package main.scala.org.huel.dataprocessing


import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.Properties

/**
 * @author lwj
 * @Description Spark Streaming 消费考勤数据并进行统计分析
 * @date 2025/06/13
 */
object KafkaAttendanceConsumer {
  def main(args: Array[String]): Unit = {
    // 1. 创建 SparkConf
    val sparkConf = new SparkConf()
      .setAppName("AttendanceConsumer")
      .setMaster("local[*]") // 本地运行模式，生产环境可以去掉
      .set("spark.streaming.stopGracefullyOnShutdown", "true")

    // 2. 创建 StreamingContext，设置批处理间隔为5秒
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    // 3. 设置日志级别为WARN，减少控制台输出信息
    ssc.sparkContext.setLogLevel("WARN")

    // 4. 配置 Kafka 消费者参数
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "43.143.125.94:9092", // Kafka 服务器地址
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "spark_attendance_consumer_group",
      "auto.offset.reset" -> "latest", // 从最新消息开始消费
      "enable.auto.commit" -> (false: java.lang.Boolean) // 手动提交偏移量
    )

    // 5. 要订阅的 Kafka 主题
    val topics = Array("attendance")

    // 6. 创建直接流
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    // 7. 处理每条消息
    stream.foreachRDD { rdd =>
      // 存储偏移量范围以便稍后提交
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      if (!rdd.isEmpty()) {
        // 输出批次信息
        println(s"========== 新批次数据 - ${java.time.LocalDateTime.now()} ==========")

        // 处理RDD中的每条记录
        rdd.foreachPartition { partitionOfRecords =>
          // 在每个执行器上创建一个数据库连接
          val connection = createConnection()

          try {
            // 准备插入原始数据语句 - 使用attendance_raw表
            val insertRawStmt = connection.prepareStatement(
              """
                |INSERT INTO attendance (class_id, student_name, course_name, student_id, score, is_absent)
                |VALUES (?, ?, ?, ?, ?, ?)
                |""".stripMargin)

            // 准备更新班级出勤统计语句
            val updateClassAttendanceStmt = connection.prepareStatement(
              """
                |INSERT INTO class_attendance (class_id, present_count, absent_count, last_updated)
                |VALUES (?, ?, ?, NOW())
                |ON DUPLICATE KEY UPDATE
                |present_count = present_count + VALUES(present_count),
                |absent_count = absent_count + VALUES(absent_count),
                |last_updated = NOW()
                |""".stripMargin)

            // 准备更新课程出勤统计语句
            val updateCourseAttendanceStmt = connection.prepareStatement(
              """
                |INSERT INTO course_attendance (course_name, present_count, absent_count, last_updated)
                |VALUES (?, ?, ?, NOW())
                |ON DUPLICATE KEY UPDATE
                |present_count = present_count + VALUES(present_count),
                |absent_count = absent_count + VALUES(absent_count),
                |last_updated = NOW()
                |""".stripMargin)

            // 准备更新总出勤统计语句
            val updateAttendanceSummaryStmt = connection.prepareStatement(
              """
                |INSERT INTO attendance_summary (total_present, total_absent, last_updated)
                |VALUES (?, ?, NOW())
                |ON DUPLICATE KEY UPDATE
                |total_present = total_present + VALUES(total_present),
                |total_absent = total_absent + VALUES(total_absent),
                |last_updated = NOW()
                |""".stripMargin)

            // 用于记录各班级统计信息
            val classStats = scala.collection.mutable.Map[String, (Int, Int)]()

            // 用于记录各课程统计信息
            val courseStats = scala.collection.mutable.Map[String, (Int, Int)]()

            // 总出勤统计
            var totalPresent = 0
            var totalAbsent = 0

            // 处理分区中的每条记录
            partitionOfRecords.foreach { record =>
              // 解析记录值
              val value = record.value()
              println(s"接收到的数据: $value")

              try {
                // 按TAB分割字段（符合你的生产者数据格式）
                val fields = value.split("\t")

                if (fields.length == 6) {
                  val classNumber = fields(0)
                  val studentName = fields(1)
                  val courseName = fields(2)
                  val studentId = fields(3)
                  val score = fields(4).toInt
                  val isAbsent = fields(5).toInt

                  // 插入原始数据
                  insertRawStmt.setString(1, classNumber)
                  insertRawStmt.setString(2, studentName)
                  insertRawStmt.setString(3, courseName)
                  insertRawStmt.setString(4, studentId)
                  insertRawStmt.setInt(5, score)
                  insertRawStmt.setInt(6, isAbsent)
                  insertRawStmt.executeUpdate()

                  // 更新班级统计信息
                  val currentClassStats = classStats.getOrElse(classNumber, (0, 0))
                  if (isAbsent == 1) {
                    classStats(classNumber) = (currentClassStats._1, currentClassStats._2 + 1) // 缺勤+1
                  } else {
                    classStats(classNumber) = (currentClassStats._1 + 1, currentClassStats._2) // 出勤+1
                  }

                  // 更新课程统计信息
                  val currentCourseStats = courseStats.getOrElse(courseName, (0, 0))
                  if (isAbsent == 1) {
                    courseStats(courseName) = (currentCourseStats._1, currentCourseStats._2 + 1) // 缺勤+1
                  } else {
                    courseStats(courseName) = (currentCourseStats._1 + 1, currentCourseStats._2) // 出勤+1
                  }

                  // 更新总出勤统计
                  if (isAbsent == 1) {
                    totalAbsent += 1
                  } else {
                    totalPresent += 1
                  }
                } else {
                  println(s"数据格式不正确: $value, 字段数量: ${fields.length}")
                  println(s"拆分后的数据: ${fields.mkString("|")}") // 打印拆分后的字段，帮助调试
                }
              } catch {
                case e: Exception =>
                  println(s"处理记录失败: ${e.getMessage}, 数据: $value")
                  e.printStackTrace()
              }
            }

            // 更新班级出勤统计
            classStats.foreach { case (classId, (present, absent)) =>
              updateClassAttendanceStmt.setString(1, classId)
              updateClassAttendanceStmt.setInt(2, present)
              updateClassAttendanceStmt.setInt(3, absent)
              updateClassAttendanceStmt.executeUpdate()
              println(s"班级: $classId, 出勤人次: $present, 缺勤人次: $absent")
            }

            // 更新课程出勤统计
            courseStats.foreach { case (courseName, (present, absent)) =>
              updateCourseAttendanceStmt.setString(1, courseName)
              updateCourseAttendanceStmt.setInt(2, present)
              updateCourseAttendanceStmt.setInt(3, absent)
              updateCourseAttendanceStmt.executeUpdate()
              println(s"课程: $courseName, 出勤人次: $present, 缺勤人次: $absent")
            }

            // 更新总出勤统计
            if (totalPresent > 0 || totalAbsent > 0) {
              updateAttendanceSummaryStmt.setInt(1, totalPresent)
              updateAttendanceSummaryStmt.setInt(2, totalAbsent)
              updateAttendanceSummaryStmt.executeUpdate()
              println(s"总出勤人次: $totalPresent, 总缺勤人次: $totalAbsent")
            }

            // 关闭预编译语句
            insertRawStmt.close()
            updateClassAttendanceStmt.close()
            updateCourseAttendanceStmt.close()
            updateAttendanceSummaryStmt.close()

          } catch {
            case e: Exception =>
              println(s"数据库操作异常: ${e.getMessage}")
              e.printStackTrace()
          } finally {
            // 关闭连接
            if (connection != null) {
              connection.close()
            }
          }
        }

        // 手动提交偏移量
        stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
      }
    }

    // 8. 启动 StreamingContext
    ssc.start()
    ssc.awaitTermination()
  }

  /**
   * 创建MySQL数据库连接
   */
  def createConnection(): Connection = {
    // 加载数据库驱动
    Class.forName("com.mysql.cj.jdbc.Driver")

    // 数据库连接信息
    val url = "jdbc:mysql://43.143.125.94:3306/attendance_new?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai"
    val user = "root"
    val password = "Lwj378$$"

    // 创建连接
    DriverManager.getConnection(url, user, password)
  }
}